package com.walala.logic.mq;

import com.alibaba.fastjson.JSON;
import com.walala.logic.cache.ServerCache;
import com.walala.logic.common.Constant;
import com.walala.logic.handler.MsgHandler;
import com.walala.logic.pojo.msg.MsgProtoc;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author KayLi
 */
public class KafkaConsumerClient {

    private Logger logger = LoggerFactory.getLogger(KafkaConsumerClient.class);
    
    private static KafkaConsumerClient instance = new KafkaConsumerClient();

    private KafkaConsumerClient(){}

    public static KafkaConsumerClient getInstance() {
        return instance;
    }

    private AtomicInteger count = new AtomicInteger(0);

    private KafkaConsumer consumer;

    public void start(){
        Properties props = new Properties();

        props.put("bootstrap.servers", "192.168.56.101:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        consumer = new KafkaConsumer(props);
        consumer.subscribe(Arrays.asList(Constant.Kafka_Notice_Msg_Topic));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord record : records) {
                logger.info("消费到消息，key:[{}],value:[{}]", record.key(), record.value());
                dispathcer(String.valueOf(record.value()));
            }
        }
    }

    private void dispathcer(String value) {

        MsgProtoc msgProtoc = JSON.parseObject(value, MsgProtoc.class);
        short cmd = msgProtoc.getCmd();
        MsgHandler handler = ServerCache.getHandler(cmd);
        if(handler != null){
            handler.process(msgProtoc);
        }


    }


}
